package com.we.risk.phoneRecord.stage2.dimphone;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.contrib.streaming.state.RocksDBStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.KeyedBroadcastProcessFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkKafkaPartitioner;
import org.apache.flink.streaming.util.serialization.KeyedSerializationSchema;
import org.apache.flink.util.Collector;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.we.flink.utils.WeKafkaPropertyReader;
import com.we.risk.phoneRecord.common.PhoneRecordDataUtils;
import com.we.risk.phoneRecord.common.Stage2GlobalSetting;
import org.apache.commons.io.FileUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.File;
import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.*;

@Deprecated
public class DimPhoneRecordLabelBroadCast {
    public static final String RELEASEPROP =
            "risk/phonerecord/stage2/dim_phone_kfk_prod.properties";
    public static final String PROPSETS = "risk/phonerecord/phonerecord.properties";
    public static Logger LOG = LoggerFactory.getLogger(DimPhoneRecordLabelBroadCast.class);

    //    public static final String KFKBROKERHOSTBAK =
    // "10.10.13.22:9092,10.10.13.23:9092,10.10.13.24:9092";
    //    public static final String KFKBAKTOPIC = "kafka.topic.data.record";

    public static final String KFKBROKERHOSTBAK =
            "10.10.15.25:9092,10.10.15.26:9092,10.10.15.27:9092";
    public static final String KFKBAKTOPIC = "AAA_risk_phone_record_label_stage2_source_test2";
    public static final String KFKBAKTOPICGRPID = "DimPhoneRecordLabel_grp";

    public static final String MOBILE_AREA_BROADCAST_DESC = "phonerecord_mob_area_broad_cast_desc";
    public static final int MOBILE_AREA_BROADCAST_WAIT_SEC = 15 * 1000;
    public static final String SPLITWORD = "#";
    public static final String DATEFORMAT = "yyyy-MM-dd";
    public static final String MSGTYPE = "HFQ_UPLOAD_PHONE_CALL_RECORD";
    public static final String INPUTKEYBY = "userKey";
    public static final String OUTPUTKEY = "user_key";
    public static final String TIMESTAMP = "timestamp";
    public static final String JOBID = "jobID";
    public static final String PHONECALLLIST = "phoneCallList";
    public static final String REMOVEJSONKEY = "numberType";
    public static final Long KFKMSGMAXSIZE = 50 * 1024 * 1024L;

    public static void main(String[] args) throws IOException {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        try {
            WeKafkaPropertyReader paramReader = WeKafkaPropertyReader.init(RELEASEPROP);
            /** RocksDB */
            env.setStateBackend(new RocksDBStateBackend(paramReader.getRocksDBBackendUrl()));
            /** checkpoint configure */
            CheckpointConfig ckConf = env.getCheckpointConfig();
            ckConf.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
            ckConf.setCheckpointInterval(10 * 60 * 1000); // ms
            ckConf.setCheckpointTimeout(60 * 60 * 1000);
            ckConf.setMaxConcurrentCheckpoints(1);
            ckConf.setMinPauseBetweenCheckpoints(500);
            ckConf.enableExternalizedCheckpoints(
                    CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);

            ckConf.enableUnalignedCheckpoints();

            /** Consumer Kafka */
            String srcKafkaTopic = paramReader.getTmpKfkTopic();
            String srcKafkaBootStrapServer = paramReader.getTmpKfkBootStrapServer();
            String srckfkGrupId = paramReader.getTmpKfkGroupId();

            Properties consumProp = new Properties();
            consumProp.setProperty("group.id", srckfkGrupId);
            consumProp.setProperty(
                    "key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
            consumProp.setProperty(
                    "value.deserializer",
                    "org.apache.kafka.common.serialization.StringDeserializer");

            /** set offset to latest */
            KafkaSource<String> kfkSource =
                    KafkaSource.<String>builder()
                            .setBootstrapServers(srcKafkaBootStrapServer)
                            .setTopics(srcKafkaTopic)
                            .setStartingOffsets(OffsetsInitializer.earliest())
                            .setProperties(consumProp)
                            .setValueOnlyDeserializer(new SimpleStringSchema())
                            .build();

            consumProp.setProperty("group.id", KFKBAKTOPICGRPID);
            KafkaSource<String> kfkBakSource =
                    KafkaSource.<String>builder()
                            .setBootstrapServers(KFKBROKERHOSTBAK)
                            .setTopics(KFKBAKTOPIC)
                            .setStartingOffsets(OffsetsInitializer.earliest())
                            .setProperties(consumProp)
                            .setValueOnlyDeserializer(new SimpleStringSchema())
                            .build();

            /** Dim Mobile Area Data */
            consumProp.setProperty("group.id", Stage2GlobalSetting.KFKMOBAREATOPICGRPID);
            KafkaSource<String> kfkMobAreaDimSource =
                    KafkaSource.<String>builder()
                            .setBootstrapServers(Stage2GlobalSetting.MOBILE_AREA_KFK_BOOTADD)
                            .setTopics(Stage2GlobalSetting.MOBILE_AREA_KFK_TOPIC)
                            .setStartingOffsets(OffsetsInitializer.earliest())
                            .setProperties(consumProp)
                            .setValueOnlyDeserializer(new SimpleStringSchema())
                            .build();
            /** distributed cache for Sougou+Cuishou Tag + mob_area_clean Data */
            env.registerCachedFile(
                    Stage2GlobalSetting.V1CLEANDATA, Stage2GlobalSetting.V1CLEANCATCHFILENAME);
            env.registerCachedFile(
                    Stage2GlobalSetting.SGCSTAGDATA, Stage2GlobalSetting.SGCSCATCHFILENAME);

            SingleOutputStreamOperator<String> input =
                    env.fromSource(kfkSource, WatermarkStrategy.noWatermarks(), "PhoneRecord")
                            .uid("kfk-PhoneRecordLabel-source");
            SingleOutputStreamOperator<String> inputBak =
                    env.fromSource(
                                    kfkBakSource,
                                    WatermarkStrategy.noWatermarks(),
                                    "PhoneRecord-bak")
                            .uid("kfk-PhoneRecordLabel-bak-source");
            SingleOutputStreamOperator<String> dimInput =
                    env.fromSource(
                                    kfkMobAreaDimSource,
                                    WatermarkStrategy.noWatermarks(),
                                    "mobile_area_dim_data")
                            .uid("mobile_area_dim_data");

            DataStream<String> unionInput = input.union(inputBak);
            MapStateDescriptor<String, String> broadcastDescriptor =
                    new MapStateDescriptor<>(
                            MOBILE_AREA_BROADCAST_DESC, String.class, String.class);
            BroadcastStream<String> broadcastStreamInput = dimInput.broadcast(broadcastDescriptor);

            SingleOutputStreamOperator<String> out =
                    unionInput
                            .flatMap(new PhoneRecordDataFilter())
                            .keyBy(
                                    new KeySelector<String, String>() {
                                        @Override
                                        public String getKey(String value) throws Exception {
                                            JSONObject jsonObject = JSONObject.parseObject(value);
                                            return jsonObject.getString(OUTPUTKEY);
                                        }
                                    })
                            .connect(broadcastStreamInput)
                            .process(new MobileAreaProcessFunc());

            /** sink to Kafka */
            String sinkkfkTopic = paramReader.getKfkTopic();
            Properties sinkProp = new Properties();
            sinkProp.setProperty("bootstrap.servers", paramReader.getKfkBootStrapServer());
            sinkProp.setProperty("max.request.size", String.valueOf(KFKMSGMAXSIZE));
            int sinkkfkPartitions = paramReader.getKfkPartitions();

            FlinkKafkaProducer<String> kafkaProducer =
                    new FlinkKafkaProducer<String>(
                            sinkkfkTopic,
                            new WeKafkaKeyedSerializationSchema(),
                            sinkProp,
                            Optional.of(new WeKafkaCustomPartitioner()));

            out.addSink(kafkaProducer).setParallelism(sinkkfkPartitions);

            env.execute(DimPhoneRecordLabelBroadCast.class.toString());
        } catch (Exception e) {
            System.out.println("Exception: " + e.getMessage());
            LOG.error("Exception: " + e.getMessage());
            e.printStackTrace();
        }
    }

    private static class PhoneRecordDataFilter extends RichFlatMapFunction<String, String> {
        @Override
        public void flatMap(String input, Collector<String> out) throws Exception {
            try {
                if (input.contains(MSGTYPE)) {
                    JSONObject inputJson = JSONObject.parseObject(input);
                    if (inputJson != null && inputJson.getString("type").equals(MSGTYPE)) {
                        JSONObject dataJson = inputJson.getJSONObject("dataJson");
                        Long captureTime = dataJson.getLong(TIMESTAMP);
                        JSONObject job = dataJson.getJSONObject("job");
                        String jobID = job.getString(JOBID);

                        String userKey = dataJson.getString(INPUTKEYBY);
                        JSONObject data = dataJson.getJSONObject("data");
                        JSONArray calls = data.getJSONArray(PHONECALLLIST);

                        if (userKey != null) {
                            JSONObject outJson = new JSONObject();
                            outJson.put(OUTPUTKEY, userKey);

                            outJson.put(TIMESTAMP, captureTime);
                            outJson.put(JOBID, jobID);
                            outJson.put(PHONECALLLIST, calls);
                            out.collect(outJson.toString());
                        }
                    }
                }
            } catch (Exception e) {
                LOG.error(e.getMessage());
                LOG.error("flatMap1 error input: " + input);
                e.printStackTrace();
            }
        }
    }

    private static class MobileAreaProcessFunc
            extends KeyedBroadcastProcessFunction<String, String, String, String> {
        public static final MapStateDescriptor<String, String> mobAreaBrodcstdesc =
                new MapStateDescriptor<>(MOBILE_AREA_BROADCAST_DESC, String.class, String.class);

        MapState<String, String> mobAreaWaitListMState;
        SimpleDateFormat dateFormat;
        long startTime = 0L;
        //        long bStateSize = 0L;
        HashMap<String, String> mapV1Clean = new HashMap<>();
        HashMap<String, String> sgcsTagMap = new HashMap<>();

        @Override
        public void open(Configuration parameters) throws Exception {
            mobAreaWaitListMState =
                    getRuntimeContext()
                            .getMapState(
                                    new MapStateDescriptor<String, String>(
                                            "MobAreaWaitListMState", String.class, String.class));
            startTime = System.currentTimeMillis();
            dateFormat = new SimpleDateFormat(DATEFORMAT);
            initV1CleanMap();
            initSgCsTagMap();
        }

        @Override
        public void processElement(String value, ReadOnlyContext ctx, Collector<String> out)
                throws Exception {
            try {
                long timediff = (ctx.timerService().currentProcessingTime() - startTime);
                ReadOnlyBroadcastState<String, String> readBState =
                        ctx.getBroadcastState(mobAreaBrodcstdesc);
                if (timediff >= MOBILE_AREA_BROADCAST_WAIT_SEC) {
                    // real process element
                    if (readBState != null) {
                        if (!mobAreaWaitListMState.isEmpty()) {
                            mobAreaWaitListMState.clear();
                        }
                        ProcessData(value, readBState, out);
                    }
                } else {
                    LOG.debug(
                            "debug MobileAreaProcessFunc processElement addinto wait List value: "
                                    + value);
                    mobAreaWaitListMState.put(OUTPUTKEY, value);
                    long now = ctx.timerService().currentProcessingTime();
                    ctx.timerService()
                            .registerProcessingTimeTimer(now + MOBILE_AREA_BROADCAST_WAIT_SEC);
                }

            } catch (Exception e) {
                LOG.error("MobileAreaProcessFunc processElement exception: " + e.toString());
                LOG.error("MobileAreaProcessFunc processElement error input: " + value);
                e.printStackTrace();
            }
        }

        /**
         * {"id":17278,"paragraph":"1317276","province_simple":"广东","city_county_simple":"河源","isp":"中国联通"}
         */
        @Override
        public void processBroadcastElement(String value, Context ctx, Collector<String> out)
                throws Exception {
            try {
                BroadcastState<String, String> writeBState =
                        ctx.getBroadcastState(mobAreaBrodcstdesc);
                if (value != null) {
                    JSONObject inputJson = JSONObject.parseObject(value);
                    String mobile = inputJson.getString("paragraph");
                    String province = inputJson.getString("province_simple");
                    String city = inputJson.getString("city_county_simple");

                    if (province == null || province.equals("")) {
                        province = "null";
                    }

                    if (city == null || city.equals("")) {
                        city = "null";
                    }

                    writeBState.put(mobile, province + SPLITWORD + city);
                    //                    bStateSize += 1;
                }
            } catch (Exception e) {
                LOG.error(
                        "MobileAreaProcessFunc processBroadcastElement exception: " + e.toString());
                LOG.error("MobileAreaProcessFunc processBroadcastElement error input: " + value);
                e.printStackTrace();
            }
        }

        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out)
                throws Exception {
            super.onTimer(timestamp, ctx, out);
            try {
                LOG.debug(
                        "debug MobileAreaProcessFunc onTimer triggered isEmpty: "
                                + mobAreaWaitListMState.isEmpty());
                if (!mobAreaWaitListMState.isEmpty()) {
                    String waitMsg = mobAreaWaitListMState.get(OUTPUTKEY);
                    ReadOnlyBroadcastState<String, String> readBState =
                            ctx.getBroadcastState(mobAreaBrodcstdesc);
                    ProcessData(waitMsg, readBState, out);
                    mobAreaWaitListMState.clear();
                }
                ctx.timerService().deleteProcessingTimeTimer(timestamp);
            } catch (Exception e) {
                LOG.error("MobileAreaProcessFunc onTimer exception: " + e.toString());
                e.printStackTrace();
            }
        }

        private void ProcessData(
                String value,
                ReadOnlyBroadcastState<String, String> readBState,
                Collector<String> out) {
            try {
                JSONObject inputJson = JSONObject.parseObject(value);
                JSONArray phoneCallList = inputJson.getJSONArray("phoneCallList");
                int callSize = phoneCallList.size();
                for (int i = 0; i < callSize; i++) {
                    JSONObject callJson = (JSONObject) phoneCallList.get(i);
                    if (callJson.containsKey(REMOVEJSONKEY)) {
                        callJson.remove(REMOVEJSONKEY);
                    }

                    String number = callJson.getString("number");
                    LOG.debug(
                            "ProcessData input number:"
                                    + number
                                    + " format_number: "
                                    + PhoneRecordDataUtils.forMattedPhoneNumber(number));
                    number = PhoneRecordDataUtils.forMattedPhoneNumber(number);
                    callJson.put("number", number);

                    String sgtag = null;
                    String coname = null;
                    String sgcsTag = sgcsTagMap.get(number);
                    if (sgcsTag != null) {
                        String[] splits = sgcsTag.split(SPLITWORD);
                        if (splits.length == 2) {
                            sgtag = splits[0];
                            coname = splits[1];
                        } else {
                            LOG.error("ProcessData sgcsTagMap sgcsTag err: " + sgcsTag);
                        }
                    }
                    //                    else {
                    //                        LOG.warn("ProcessData sgcsTagMap warn number: "
                    // +number);
                    //                    }
                    callJson.put("sgtag", sgtag);
                    callJson.put("coname", coname);

                    String mob_prov = null;
                    String mob_city = null;
                    String[] addrs = null;
                    if (!number.equals("") && number.length() == 11) {
                        String subNumber = number.substring(0, 7);
                        addrs = MapMobArea(subNumber, readBState);
                        if (addrs != null) {
                            LOG.debug(
                                    "ProcessData MapMobArea returns prov:"
                                            + addrs[0]
                                            + " city: "
                                            + addrs[1]);
                            callJson.put("mob_prov", addrs[0]);
                            callJson.put("mob_city", addrs[1]);
                        }
                    }

                    phoneCallList.set(i, callJson);
                }

                out.collect(inputJson.toString());

            } catch (Exception e) {
                LOG.error("ProcessData exception: " + e.toString());
                LOG.error(
                        "ProcessData exception input: "
                                + (value.length() > 100 ? value.substring(0, 100) : value));
                e.printStackTrace();
            }
        }

        private void initV1CleanMap() throws IOException {
            File v1CleanFile =
                    getRuntimeContext()
                            .getDistributedCache()
                            .getFile(Stage2GlobalSetting.V1CLEANCATCHFILENAME);
            List<String> v1CleanLines = FileUtils.readLines(v1CleanFile);
            for (String line : v1CleanLines) {
                String[] splits = line.split(",");
                if (splits.length == 4) {
                    String new_prov = splits[0];
                    String new_city = splits[1];
                    String raw_prov = splits[2];
                    String raw_city = splits[3];
                    String mapKey = raw_prov + SPLITWORD + raw_city;
                    String mapVal = new_prov + SPLITWORD + new_city;
                    mapV1Clean.put(mapKey, mapVal);
                } else {
                    LOG.error("DimPhoneRecordLabel initV1CleanMap error read line: " + line);
                }
            }
        }

        private void initSgCsTagMap() throws IOException {
            File sgTagFile =
                    getRuntimeContext()
                            .getDistributedCache()
                            .getFile(Stage2GlobalSetting.SGCSCATCHFILENAME);
            List<String> sgTagLines = FileUtils.readLines(sgTagFile);
            for (String line : sgTagLines) {
                String[] splits = line.split(",");
                if (splits.length == 3) {
                    String phone_num = splits[0];
                    String sgTag = splits[1];
                    String coName = splits[2];
                    String mapKey = phone_num;
                    String mapVal = sgTag + SPLITWORD + coName;
                    sgcsTagMap.put(mapKey, mapVal);
                } else {
                    LOG.error("DimPhoneRecordLabel initSgCsTagMap error read line: " + line);
                }
            }
        }

        /** 返回映射完全的地址 */
        private String[] MapMobArea(
                String mobile, ReadOnlyBroadcastState<String, String> readBState) throws Exception {
            String res = readBState.get(mobile);
            String[] newAddrs = new String[2];
            if (res != null) {
                LOG.debug("MapMobArea get res: " + res);
                String[] splits = res.split(SPLITWORD);
                if (splits.length == 2) {
                    String mob_prov = (splits[0].equals("null") ? "" : splits[0]);
                    String mob_city = (splits[1].equals("null") ? "" : splits[1]);

                    String trans_key = mob_prov + SPLITWORD + mob_city;
                    String trans_val = mapV1Clean.get(trans_key);
                    LOG.debug("MapMobArea get trans_val: " + trans_val);
                    if (trans_val != null) {
                        newAddrs = trans_val.split(SPLITWORD);
                        if (newAddrs.length == 2) {
                            LOG.debug(
                                    "MapMobArea get prov: "
                                            + newAddrs[0]
                                            + " city: "
                                            + newAddrs[1]);
                            return newAddrs;
                        } else {
                            LOG.warn(
                                    "DimPhoneRecordLabel MapMobArea V1Clean data error: "
                                            + trans_val);
                        }
                    } else {
                        LOG.warn("DimPhoneRecordLabel MapMobArea err trans_key: " + trans_key);
                    }

                    //                    LOG.warn("Debug UserJoinMobileAreaProcessFunc ProcessData
                    // bstateSize: " + bStateSize + " res: " +
                    // inputJson.toString(SerializerFeature.WriteMapNullValue));
                } else {
                    LOG.warn("MobileAreaProcessFunc MapMobArea error res: " + res);
                }
            } else {
                LOG.debug("MapMobArea readBstate cant find mob : " + mobile);
            }
            return null;
        }
    }

    private static class WeKafkaKeyedSerializationSchema
            implements KeyedSerializationSchema<String> {
        @Override
        public byte[] serializeKey(String element) {
            JSONObject jsonObject = JSONObject.parseObject(element);
            String keyby = jsonObject.getString(OUTPUTKEY);
            return keyby.getBytes();
        }

        @Override
        public byte[] serializeValue(String element) {
            return element.getBytes();
        }

        @Override
        public String getTargetTopic(String element) {
            return null;
        }
    }

    private static class WeKafkaCustomPartitioner extends FlinkKafkaPartitioner<String> {

        @Override
        public int partition(
                String record, byte[] key, byte[] value, String targetTopic, int[] partitions) {
            int partition = Math.abs(new String(key).hashCode() % partitions.length);
            if (LOG.isDebugEnabled()) {
                LOG.info(
                        " partitions: "
                                + partitions.length
                                + " partition: "
                                + partition
                                + " key: "
                                + new String(key));
            }
            return partition;
        }
    }
}
